package com.dataseek.iot.mqtt.service.impl;

import com.dataseek.iot.mqtt.callback.DefaultCallback;
import com.dataseek.iot.mqtt.service.MqttGateway;
import lombok.NonNull;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.Objects;

/**
 * @author cc
 * @Title: 默认MQTT网关
 * @Description: 默认MQTT网关
 * @date 2021/9/255:13 下午
 * @since 1.0
 */
@Service
public class DefaultMqttGateway implements MqttGateway {

    private MqttClient client;

    @Value("${mqtt.brokerUrl}")
    private String url;

    @Value("${mqtt.publisher.clientId}")
    private String clientId;

    @Value("${mqtt.username}")
    private String username;

    @Value("${mqtt.password}")
    private String password;

    @Value("${mqtt.publisher.defaultTopic}")
    private String publisherTopic;

    @Value("${mqtt.subscriber.defaultTopic}")
    private String subscriberTopic;

    @Autowired
    private DefaultCallback defaultCallback;


    private void setMqttClient() throws MqttException {
        client = new MqttClient(url, clientId);
        MqttConnectOptions options = mqttConnectOptions(username, password);
        client.setCallback(defaultCallback);
        client.connect(options);
    }

    /**
     * 设置mqtt客户端
     * @param username
     * @param password
     * @param callback
     */
    public void setMqttClient(@NonNull String username, @NonNull String password, MqttCallback callback) throws MqttException {
        client = new MqttClient(url, clientId);
        MqttConnectOptions options = mqttConnectOptions(username, password);
        client.setCallback(callback);
        client.connect(options);
    }


    //TODO 临时用法
    public void initClient() throws MqttException {
        if(Objects.isNull(client)){
            setMqttClient();
        }
    }


    /**
     * MQTT连接参数设置
     * @param username
     * @param password
     * @return
     * @throws MqttException
     */
    private MqttConnectOptions mqttConnectOptions(String username, String password) throws MqttException {
        MqttConnectOptions options = new MqttConnectOptions();
        if(username!=null&&!username.isEmpty()){
            options.setUserName(username);
        }
        if(password!=null&&!password.isEmpty()){
            options.setPassword(password.toCharArray());
        }
        options.setConnectionTimeout(10);
        options.setAutomaticReconnect(true);
        options.setCleanSession(true);
        return options;
    }



    @Override
    public void sendToMqtt(String message) throws MqttException {
        initClient();
        MqttMessage msg = new MqttMessage();
        msg.setPayload(message.getBytes(StandardCharsets.UTF_8));
        MqttTopic mqttTopic = client.getTopic(publisherTopic);
        MqttDeliveryToken token = mqttTopic.publish(msg);
        token.waitForCompletion();

    }

    @Override
    public void sendToMqtt(String topic, String msg) throws MqttException {
        initClient();
        MqttMessage message = new MqttMessage();
        message.setPayload(msg.getBytes(StandardCharsets.UTF_8));
        MqttTopic mqttTopic = client.getTopic(topic);
        MqttDeliveryToken token = mqttTopic.publish(message);
        token.waitForCompletion();
    }

    @Override
    public void sendToMqtt(String topic, int qos, String msg) throws MqttException {
        initClient();
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(qos);
        mqttMessage.setPayload(msg.getBytes());
        MqttTopic mqttTopic = client.getTopic(topic);
        MqttDeliveryToken token = null;
        token = mqttTopic.publish(mqttMessage);
        token.waitForCompletion();
    }

    /**
     * 关闭MQTT连接
     */
    public void close(){
        try {
            client.close();
            client.disconnect();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 订阅默认主题消息
     * @throws MqttException
     */
    public void sub() throws MqttException {
        client.setCallback(defaultCallback);
        client.subscribe(subscriberTopic);
    }
    /**
     * 订阅消息
     * @param topic
     * @throws MqttException
     */
    public void sub(String topic) throws MqttException {
        client.subscribe(topic);
    }

    /**
     * 订阅多主题
     * @param topics
     * @throws MqttException
     */
    public void sub(String... topics) throws MqttException {
        client.subscribe(topics);
    }

}
